package org.zebra.redisson.queue;

import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RFuture;
import org.zebra.redisson.util.RedissonUtil;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 延迟队列
 *
 * @author yue
 */
public final class DelayedQueue<V> {

    private static final Map<String, DelayedQueue> delayedQueueCache = new ConcurrentHashMap<>();

    /**
     * 发送消息队列
     */
    private RDelayedQueue<V> producerQueue;
    /**
     * 消费消息队列
     */
    private RBlockingQueue<V> consumerQueue;

    private DelayedQueue(RBlockingQueue<V> consumerQueue, RDelayedQueue<V> producerQueue) {
        this.consumerQueue = consumerQueue;
        this.producerQueue = producerQueue;
    }

    public RDelayedQueue<V> getProducerQueue() {
        return producerQueue;
    }

    public RBlockingQueue<V> getConsumerQueue() {
        return consumerQueue;
    }

    /**
     * 获得延迟队列
     *
     * @param queueName 队列名
     */
    public static <V> DelayedQueue<V> get(String queueName) {
        DelayedQueue<V> delayedQueue = delayedQueueCache.get(queueName);
        if (delayedQueue == null) {
            RBlockingQueue<V> consumerQueue = RedissonUtil.getClient().getBlockingQueue(queueName);
            RDelayedQueue<V> producerQueue = RedissonUtil.getClient().getDelayedQueue(consumerQueue);
            delayedQueue = new DelayedQueue(consumerQueue, producerQueue);
            delayedQueueCache.put(queueName, delayedQueue);
        }
        return delayedQueue;
    }

    /**
     * 发送延迟消息
     *
     * @param msg      要发送的数据
     * @param delay    延迟多久
     * @param timeUnit 时间单位
     */
    public void send(V msg, long delay, TimeUnit timeUnit) {
        producerQueue.offer(msg, delay, timeUnit);
    }

    /**
     * 发送延迟消息（异步）
     *
     * @param msg      要发送的数据
     * @param delay    延迟多久
     * @param timeUnit 时间单位
     */
    public RFuture<Void> sendAsync(V msg, long delay, TimeUnit timeUnit) {
        return producerQueue.offerAsync(msg, delay, timeUnit);
    }

    /**
     * 消费一条延迟消息
     */
    public V consumer() {
        return consumerQueue.poll();
    }

    /**
     * 消费一条延迟消息（异步）
     */
    public RFuture<V> consumerAsync() {
        return consumerQueue.takeAsync();
    }

    /**
     * 持续消费延迟消息（监听延迟消息）
     *
     * @param consumer lambda实现消费逻辑
     * @return listenerId - 侦听器的id
     */
    public int consumer(Consumer<V> consumer) {
        return consumerQueue.subscribeOnElements(consumer);
    }

    /**
     * 删除延迟队列数据
     */
    public boolean remove(V data) {
        return producerQueue.remove(data);
    }

    /**
     * 销毁延迟队列 所有阻塞监听 报错
     */
    public void destroy() {
        producerQueue.destroy();
    }
}
